package org.niit.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;

// 手动提交偏移量
public class KafkaConsumerHandSync {

    public static void main(String[] args) throws InterruptedException {
        //1.配置信息
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        //指定消费者组
        props.put(ConsumerConfig.GROUP_ID_CONFIG,"G4");
        //手动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

        //将消费出来的数据进行转换 二进制---》String
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());

        //2.创建Kafka消费者
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props);

        //3.订阅组主题
        consumer.subscribe(Arrays.asList("BD2") );




        //4.开启死循环，不断的从Kafka中获取数据
        for(;;){

            //4.1 Kafka的消费者一次会拉去一批（很多数据）的数据
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            //4.2对这一批数据进行逐条遍历打印
            for (ConsumerRecord<String,String> data:records){
                String topic = data.topic();
                String value = data.value();
                long offset = data.offset();
                System.out.println("主题："+topic+",偏移量："+offset+",值："+value);

                //手动提交 同步方式
                //consumer.commitSync();

                //手动提交 异步方式
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {

                        if(e == null){
                            System.out.println("消费成功:"+map);
                        }else{
                            System.err.println(e.getMessage());
                        }

                    }
                });


                Thread.sleep(1000);
            }
        }



    }
}
